package cn.doitedu.demo2;


import cn.doitedu.beans.UserAction;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.HashMap;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/3/7
 * @Desc: 学大数据，上多易教育
 *
 * 规则id: rule-004
 * 规则条件：
 * 动态画像统计条件
 *     添加购物车事件达到 4次以上
 *     且历次添加到购物车的商品的价格平均值>80.5
 * 触发条件：
 *     发生下单事件


 * 规则id: rule-005
 * 规则条件：
 * 动态画像统计条件
 *     收藏商品事件达到 6次以上
 *     且历次收藏的商品的价格平均值>120
 * 触发条件：
 *     发生分享事件

 *
 *
 **/

public class RuleCalculator_04 implements RuleCalculator {


    ValueState<Integer> cntState;
    ValueState<Double> sumState;

    private final String ruleId = "rule-004";
    JSONObject message;

    @Override
    public void init(RuntimeContext runtimeContext) throws IOException {
        cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>(ruleId() + "-cnt", Integer.class));
        sumState = runtimeContext.getState(new ValueStateDescriptor<Double>(ruleId() + "-sum", Double.class));

        message = new JSONObject();
        message.put("rule_id",ruleId);
    }

    @Override
    public void calc(UserAction userAction, Collector<String> out) throws IOException {

        String eventId = userAction.getEvent_id();
        HashMap<String, String> properties = userAction.getProperties();


        if (eventId.equals("add_cart")) {
            cntState.update(cntState.value() == null ? 1 : cntState.value() + 1);

            double itemPrice = Double.parseDouble(properties.get("item_price"));
            sumState.update( sumState.value() == null ? itemPrice : sumState.value()+itemPrice);
        }


        if(eventId.equals("create_order")){

            Integer cnt = cntState.value();
            Double sum = sumState.value();
            if( cnt != null && sum !=null && cnt > 4 && sum/cnt > 80.5){
                message.put("user_id",userAction.getUser_id());
                message.put("hit_time",userAction.getAction_time());

                out.collect(message.toJSONString());

            }

        }
    }

    @Override
    public String ruleId() {
        return this.ruleId;
    }
}
